package com.zhengb.flink.study.datastream.transformation;

import com.zhengb.flink.study.datastream.datasource.RandomParallelSource;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * 自定义非并行 source function
 * <p>
 * DataStream -> DataStream
 *
 * @author zhengbo
 * @date 2019/12/19
 */
@Slf4j
public class FilterTransformation {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(new RandomParallelSource()).setParallelism(10)
                .filter(new FilterFunction<Long>() {
                    @Override
                    public boolean filter(Long value) throws Exception {
                        log.info("receive data:{}", value);
                        return value % 2 == 0;
                    }
                }).setParallelism(5).addSink(
                new RichSinkFunction<Long>() {
                    @Override
                    public void invoke(Long value, Context context) throws Exception {
                        log.info("sink data:{}", value);
                    }
                }
        ).setParallelism(5);

        env.execute("FilterTransformation");
    }


}
